package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	"time"
)
var (
	client sarama.SyncProducer
	kc chan *kafkaChannel
)

type kafkaChannel struct {
	Topic string
	Data string
}

func Init(addr []string,maxSize int)(err error){
	config := sarama.NewConfig()
	config.Producer.RequiredAcks=sarama.WaitForAll
	config.Producer.Partitioner=sarama.NewRandomPartitioner
	config.Producer.Return.Successes=true
	client, err = sarama.NewSyncProducer(addr, config)
	if err!=nil {
		return err
	}
	fmt.Printf("init kafka success \n")
	kc=make(chan *kafkaChannel,maxSize)
	go SendToKafka()
	return nil
}

func SendToChannel(topic,data string)  {
	kcData:=&kafkaChannel{Topic: topic,Data: data}
	kc<-kcData
}

func SendToKafka(){
	fmt.Println("kafka send func")
	for{
		select {
		case data:=<-kc:
			msg:=&sarama.ProducerMessage{}
			msg.Topic=data.Topic
			msg.Value=sarama.StringEncoder(data.Data)
			message, offset, err := client.SendMessage(msg)
			if err!=nil {
				fmt.Println("error send to kafka",err)
			}
			fmt.Printf("kafka send\n")
			fmt.Printf("send msg success topic:=%v value:=%v  message:=%v  offset:=%v",data.Topic,data.Data,message,offset)
		default:
			time.Sleep(time.Millisecond*50)
		}
	}
}